package com.ht.cdc;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import com.ht.api.db.util.RedisUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * CDC的Helper
 * @author asq
 * @createTime 2025年3月9日 13:35:23
 */
@Component
@ConfigurationProperties(prefix = "spring.kafka")
@Slf4j
public class CdcHelper{
	/** redis_key:是否运行中 */
	public static final String KEY_CDC_RUNING = "cdc:%s:runing";
	/** redis_key:是否是二次启动 */
	private static final String KEY_CDC_RELOAD = "cdc:%s:reload";
	/** redis_key:指定位置 */
	private static final String KEY_CDC_FILE = "cdc:%s:file";
	/** redis_key:指定偏移 */
	private static final String KEY_CDC_POS = "cdc:%s:pos";
	/** 已启动的任务 */
	private static Map<String, JobClient> map = new HashMap<>();
	
	private String bootstrapServers;
	
    public void setBootstrapServers(String bootstrapServers) {
		this.bootstrapServers = bootstrapServers;
	}
	
    /**
     * 启动cdc同步任务
     * @param config 数据源配置
     * @param tenantId 租户ID
     * @param appId 应用ID
     * @throws Exception
     */
	public void start(CdcConfig config, long tenantId, String appId) throws Exception {
		// 读取缓存中的信息
		String keyRuning = String.format(KEY_CDC_RUNING, appId);
		String keyReload = String.format(KEY_CDC_RELOAD, appId);
		String keyFile = String.format(KEY_CDC_FILE, appId);
		String keyPos = String.format(KEY_CDC_POS, appId);
		Boolean isRuning = RedisUtil.get(keyRuning); // 是否运行中
		Boolean isReload = RedisUtil.get(keyReload); // 是否是重新启动
		String offsetFile = RedisUtil.get(keyFile); // 指定位置
		Integer pos = RedisUtil.get(keyPos); // 指定偏移
		
		// 防止重复执行任务
		if (isRuning != null && isRuning) {
			log.warn("[cdc]任务运行中，不再重复运行，直接退出，租户ID：{}", tenantId);
			return ;
		}
		
	    // 1. 设置环境
	    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	    // 检查点配置
	    env.enableCheckpointing(3000); // 每 3 秒触发一次检查点
	    CheckpointConfig checkpointConfig = env.getCheckpointConfig();
	    checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 精确一次语义
	    checkpointConfig.setMinPauseBetweenCheckpoints(500); // 最小检查点间隔 500 毫秒
	    checkpointConfig.setCheckpointTimeout(60000); // 检查点超时时间 1 分钟
	    checkpointConfig.setMaxConcurrentCheckpoints(1); // 同一时间只允许一个检查点
	    checkpointConfig.setCheckpointStorage("file:///D:/Flink/checkpoint"); // 检查点存储路径

	    // 重启策略
	    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));

	    // 2. MySQL Source
	    Properties jdbcProperties = new Properties();
	    jdbcProperties.setProperty("connectTimeout", "30000");
	    jdbcProperties.setProperty("socketTimeout", "30000");
	    jdbcProperties.setProperty("useSSL", "false");

	    MySqlSourceBuilder<String> builder = MySqlSource.<String>builder()
	        .hostname(config.getHostname())
	        .port(config.getPort())
	        .databaseList(config.getDatabase())
	        .tableList(config.getTableList())
	        .username(config.getUsername())
	        .password(config.getPassword())
	        .deserializer(new JsonDebeziumDeserializationSchema())
	        .serverId("5400-5440") // 设置唯一的 server ID
	        .jdbcProperties(jdbcProperties);
	    if (isReload == null || !isReload) {
	    	log.info("[cdc]首次启动，全量拉取，租户ID：{}", tenantId);
	    	builder.startupOptions(StartupOptions.initial()); // 首次：从初始状态读取
	    } else if (StrUtil.isBlank(offsetFile)){
	    	log.info("[cdc]二次启动，无位置信息，拉取最新数据，租户ID：{}", tenantId);
	    	builder.startupOptions(StartupOptions.latest()); // 非首次，但又没有位置信息时：读取最新数据
	    } else {
	    	log.info("[cdc]二次启动，有位置信息，但暂时使用StartupOptions.latest()模式拉取数据，租户ID：{}，file：{}，pos：{}", 
	    			tenantId, offsetFile, pos);
	    	builder.startupOptions(StartupOptions.latest()); // 非首次，但cdc返回的位置信息不准，暂时使用此模式读取最新数据
//	    	builder.startupOptions(StartupOptions.specificOffset(offsetFile, pos)); // 非首次，有位置信息时：从指定位置及偏移量读取
	    }
	    MySqlSource<String> mySqlSource = builder.build();

	    // 3. 数据处理
	    var dataStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC")
	        .map(value -> {
	        	try {
		        	log.info("[cdc]收到原始数据：{}", value);
		        	String formatMsg = formatDebeziumLog(value, appId, keyFile, keyPos);
		        	log.info("[cdc]解析后的数据：{}", formatMsg);
		        	return formatMsg;
	        	} catch (Exception e) {
	        		log.warn("[cdc]解析数据出现异常，原始数据：{}", value);
	        		return value;
	        	}
	        })
	        .filter(Objects::nonNull);

	    // 4. 配置 Kafka Sink
	    String kafkaTopic = "CDC_" + tenantId; // 动态生成主题名称
	    Properties kafkaProps = new Properties();
	    kafkaProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); // Kafka 地址

	    KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
	        .setBootstrapServers(bootstrapServers) // Kafka 地址
	        .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
	            .setTopic(kafkaTopic) // 设置 Kafka 主题
	            .setValueSerializationSchema((value) -> value.getBytes(StandardCharsets.UTF_8)) // 将数据序列化为字节数组
	            .build())
	        .setKafkaProducerConfig(kafkaProps) // 设置 Kafka 生产者配置
	        .build();

	    // 5. 将数据写入 Kafka
	    dataStream.sinkTo(kafkaSink);
	    
	    // 保存标志位
	    RedisUtil.set(keyReload, true); // 是否是二次启动
	    RedisUtil.set(keyRuning, true); // 是否运行中

	    // 6. 执行
	    JobClient jobClient = env.executeAsync();
	    
	    // 7. 装进map，以便将来可以停止
	    map.put(appId, jobClient);
	}
	
    /**
     * 格式化 Debezium 日志
     * @param log 原始 JSON 日志字符串
     * @param appId 应用ID
     * @param keyFile redis_key:指定位置
     * @param keyPos redis_key:指定偏移
     * @return 格式化后的日志字符串
     */
    private static String formatDebeziumLog(String log, String appId, String keyFile, String keyPos) {
        // 解析原始日志为 JSONObject
        JSONObject rawJson = JSONUtil.parseObj(log);
        
		// 保存数据读取位置，以便下次启动时从指定位置继续拉取
		String sourceFile = rawJson.getByPath("source.file", String.class);
		Long sourcePos = rawJson.getByPath("source.pos", Long.class);
		if (StrUtil.isNotBlank(sourceFile)) {
			RedisUtil.set(keyFile, sourceFile); // 指定位置
		}
		if (sourcePos != null) {
			RedisUtil.set(keyPos, sourcePos); // 指定偏移
		}

        // 获取操作类型
        String op = rawJson.getByPath("op", String.class);
        if (StrUtil.isBlank(op)) {
            return null;
        }

        String operationType;
        switch (op) {
	        case "r":
	        	operationType = "read";
	        	break;
            case "c":
                operationType = "insert";
                break;
            case "u":
                operationType = "update";
                break;
            case "d":
                operationType = "delete";
                break;
            default:
                operationType = "none";
        }

        // 获取表和数据库信息
        String table = rawJson.getByPath("source.table", String.class);
        String database = rawJson.getByPath("source.db", String.class);

        // 获取变化前后的数据
        JSONObject before = rawJson.getByPath("before", JSONObject.class);
        JSONObject after = rawJson.getByPath("after", JSONObject.class);

        // 获取时间戳
        Long timestamp = rawJson.getByPath("ts_ms", Long.class);

        // 构造简洁的格式化输出
        JSONObject formattedLog = new JSONObject();
        formattedLog.set("appid", appId);
        formattedLog.set("type", operationType);
        formattedLog.set("database", database);
        formattedLog.set("table", table);
        formattedLog.set("time", timestamp);
        formattedLog.set("before", before);
        formattedLog.set("after", after);

        return formattedLog.toString();
    }
    
    /**
     * 停止任务
     * @param appId 应用ID
     */
    public void stop(String appId) {
		log.info("[cdc]准备停止CDC任务，appid：{}", appId);
    	JobClient jobClient = map.get(appId);
    	if (jobClient != null) {
    		// 停止任务
    		jobClient.cancel();
    		
			// 删除“运行中”的标志，恢复为未运行
			RedisUtil.delete(String.format(CdcHelper.KEY_CDC_RUNING, appId));
			
			// 删除容器中的任务
			map.remove(appId);
			
    		log.info("[cdc]停止CDC任务成功，appid：{}", appId);
    	} else {
    		log.warn("[cdc]停止CDC任务失败，找不到任务，appid：{}", appId);
    	}
    }
}
